<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
  "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">

<html xmlns="http://www.w3.org/1999/xhtml">
<head><title></title>
<link href="../style/ebook.css" type="text/css" rel="stylesheet">
</head>
<body>
<h1>Concepts</h1>
<p>The Airflow Platform is a tool for describing, executing, and monitoring
workflows.</p>
<div class="section" id="core-ideas">
<h2 class="sigil_not_in_toc">Core Ideas</h2>
<div class="section" id="dags">
<h3 class="sigil_not_in_toc">DAGs</h3>
<p>In Airflow, a <code class="docutils literal notranslate"><span class="pre">DAG</span></code> &#x2013; or a Directed Acyclic Graph &#x2013; is a collection of all
the tasks you want to run, organized in a way that reflects their relationships
and dependencies.</p>
<p>For example, a simple DAG could consist of three tasks: A, B, and C. It could
say that A has to run successfully before B can run, but C can run anytime. It
could say that task A times out after 5 minutes, and B can be restarted up to 5
times in case it fails. It might also say that the workflow will run every night
at 10pm, but shouldn&#x2019;t start until a certain date.</p>
<p>In this way, a DAG describes <em>how</em> you want to carry out your workflow; but
notice that we haven&#x2019;t said anything about <em>what</em> we actually want to do! A, B,
and C could be anything. Maybe A prepares data for B to analyze while C sends an
email. Or perhaps A monitors your location so B can open your garage door while
C turns on your house lights. The important thing is that the DAG isn&#x2019;t
concerned with what its constituent tasks do; its job is to make sure that
whatever they do happens at the right time, or in the right order, or with the
right handling of any unexpected issues.</p>
<p>DAGs are defined in standard Python files that are placed in Airflow&#x2019;s
<code class="docutils literal notranslate"><span class="pre">DAG_FOLDER</span></code>. Airflow will execute the code in each file to dynamically build
the <code class="docutils literal notranslate"><span class="pre">DAG</span></code> objects. You can have as many DAGs as you want, each describing an
arbitrary number of tasks. In general, each one should correspond to a single
logical workflow.</p>
<div class="admonition note">
<p class="first admonition-title">Note</p>
<p class="last">When searching for DAGs, Airflow will only consider files where the string
&#x201C;airflow&#x201D; and &#x201C;DAG&#x201D; both appear in the contents of the <code class="docutils literal notranslate"><span class="pre">.py</span></code> file.</p>
</div>
<div class="section" id="scope">
<h4 class="sigil_not_in_toc">Scope</h4>
<p>Airflow will load any <code class="docutils literal notranslate"><span class="pre">DAG</span></code> object it can import from a DAGfile. Critically,
that means the DAG must appear in <code class="docutils literal notranslate"><span class="pre">globals()</span></code>. Consider the following two
DAGs. Only <code class="docutils literal notranslate"><span class="pre">dag_1</span></code> will be loaded; the other one only appears in a local
scope.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">dag_1</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;this_dag_will_be_discovered&apos;</span><span class="p">)</span>

<span class="k">def</span> <span class="nf">my_function</span><span class="p">():</span>
    <span class="n">dag_2</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;but_this_dag_will_not&apos;</span><span class="p">)</span>

<span class="n">my_function</span><span class="p">()</span>
</pre>
</div>
</div>
<p>Sometimes this can be put to good use. For example, a common pattern with
<code class="docutils literal notranslate"><span class="pre">SubDagOperator</span></code> is to define the subdag inside a function so that Airflow
doesn&#x2019;t try to load it as a standalone DAG.</p>
</div>
<div class="section" id="default-arguments">
<h4 class="sigil_not_in_toc">Default Arguments</h4>
<p>If a dictionary of <code class="docutils literal notranslate"><span class="pre">default_args</span></code> is passed to a DAG, it will apply them to
any of its operators. This makes it easy to apply a common parameter to many operators without having to type it many times.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">default_args</span> <span class="o">=</span> <span class="p">{</span>
    <span class="s1">&apos;start_date&apos;</span><span class="p">:</span> <span class="n">datetime</span><span class="p">(</span><span class="mi">2016</span><span class="p">,</span> <span class="mi">1</span><span class="p">,</span> <span class="mi">1</span><span class="p">),</span>
    <span class="s1">&apos;owner&apos;</span><span class="p">:</span> <span class="s1">&apos;Airflow&apos;</span>
<span class="p">}</span>

<span class="n">dag</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;my_dag&apos;</span><span class="p">,</span> <span class="n">default_args</span><span class="o">=</span><span class="n">default_args</span><span class="p">)</span>
<span class="n">op</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;dummy&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>
<span class="nb">print</span><span class="p">(</span><span class="n">op</span><span class="o">.</span><span class="n">owner</span><span class="p">)</span> <span class="c1"># Airflow</span>
</pre>
</div>
</div>
</div>
<div class="section" id="context-manager">
<h4 class="sigil_not_in_toc">Context Manager</h4>
<p><em>Added in Airflow 1.8</em></p>
<p>DAGs can be used as context managers to automatically assign new operators to that DAG.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="k">with</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;my_dag&apos;</span><span class="p">,</span> <span class="n">start_date</span><span class="o">=</span><span class="n">datetime</span><span class="p">(</span><span class="mi">2016</span><span class="p">,</span> <span class="mi">1</span><span class="p">,</span> <span class="mi">1</span><span class="p">))</span> <span class="k">as</span> <span class="n">dag</span><span class="p">:</span>
    <span class="n">op</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="s1">&apos;op&apos;</span><span class="p">)</span>

<span class="n">op</span><span class="o">.</span><span class="n">dag</span> <span class="ow">is</span> <span class="n">dag</span> <span class="c1"># True</span>
</pre>
</div>
</div>
</div>
</div>
<div class="section" id="operators">
<span id="concepts-operators"></span><h3 class="sigil_not_in_toc">Operators</h3>
<p>While DAGs describe <em>how</em> to run a workflow, <code class="docutils literal notranslate"><span class="pre">Operators</span></code> determine what
actually gets done.</p>
<p>An operator describes a single task in a workflow. Operators are usually (but
not always) atomic, meaning they can stand on their own and don&#x2019;t need to share
resources with any other operators. The DAG will make sure that operators run in
the correct certain order; other than those dependencies, operators generally
run independently. In fact, they may run on two completely different machines.</p>
<p>This is a subtle but very important point: in general, if two operators need to
share information, like a filename or small amount of data, you should consider
combining them into a single operator. If it absolutely can&#x2019;t be avoided,
Airflow does have a feature for operator cross-communication called XCom that is
described elsewhere in this document.</p>
<p>Airflow provides operators for many common tasks, including:</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">BashOperator</span></code> - executes a bash command</li>
<li><code class="docutils literal notranslate"><span class="pre">PythonOperator</span></code> - calls an arbitrary Python function</li>
<li><code class="docutils literal notranslate"><span class="pre">EmailOperator</span></code> - sends an email</li>
<li><code class="docutils literal notranslate"><span class="pre">SimpleHttpOperator</span></code> - sends an HTTP request</li>
<li><code class="docutils literal notranslate"><span class="pre">MySqlOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">SqliteOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">PostgresOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">MsSqlOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">OracleOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">JdbcOperator</span></code>, etc. - executes a SQL command</li>
<li><code class="docutils literal notranslate"><span class="pre">Sensor</span></code> - waits for a certain time, file, database row, S3 key, etc&#x2026;</li>
</ul>
<p>In addition to these basic building blocks, there are many more specific
operators: <code class="docutils literal notranslate"><span class="pre">DockerOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">HiveOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">S3FileTransformOperator</span></code>,
<code class="docutils literal notranslate"><span class="pre">PrestoToMysqlOperator</span></code>, <code class="docutils literal notranslate"><span class="pre">SlackOperator</span></code>&#x2026; you get the idea!</p>
<p>The <code class="docutils literal notranslate"><span class="pre">airflow/contrib/</span></code> directory contains yet more operators built by the
community. These operators aren&#x2019;t always as complete or well-tested as those in
the main distribution, but allow users to more easily add new functionality to
the platform.</p>
<p>Operators are only loaded by Airflow if they are assigned to a DAG.</p>
<p>See <a class="reference internal" href="howto/operator.html"><span class="doc">Using Operators</span></a> for how to use Airflow operators.</p>
<div class="section" id="dag-assignment">
<h4 class="sigil_not_in_toc">DAG Assignment</h4>
<p><em>Added in Airflow 1.8</em></p>
<p>Operators do not have to be assigned to DAGs immediately (previously <code class="docutils literal notranslate"><span class="pre">dag</span></code> was
a required argument). However, once an operator is assigned to a DAG, it can not
be transferred or unassigned. DAG assignment can be done explicitly when the
operator is created, through deferred assignment, or even inferred from other
operators.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">dag</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;my_dag&apos;</span><span class="p">,</span> <span class="n">start_date</span><span class="o">=</span><span class="n">datetime</span><span class="p">(</span><span class="mi">2016</span><span class="p">,</span> <span class="mi">1</span><span class="p">,</span> <span class="mi">1</span><span class="p">))</span>

<span class="c1"># sets the DAG explicitly</span>
<span class="n">explicit_op</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;op1&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>

<span class="c1"># deferred DAG assignment</span>
<span class="n">deferred_op</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;op2&apos;</span><span class="p">)</span>
<span class="n">deferred_op</span><span class="o">.</span><span class="n">dag</span> <span class="o">=</span> <span class="n">dag</span>

<span class="c1"># inferred DAG assignment (linked operators must be in the same DAG)</span>
<span class="n">inferred_op</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;op3&apos;</span><span class="p">)</span>
<span class="n">inferred_op</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">(</span><span class="n">deferred_op</span><span class="p">)</span>
</pre>
</div>
</div>
</div>
<div class="section" id="bitshift-composition">
<h4 class="sigil_not_in_toc">Bitshift Composition</h4>
<p><em>Added in Airflow 1.8</em></p>
<p>Traditionally, operator relationships are set with the <code class="docutils literal notranslate"><span class="pre">set_upstream()</span></code> and
<code class="docutils literal notranslate"><span class="pre">set_downstream()</span></code> methods. In Airflow 1.8, this can be done with the Python
bitshift operators <code class="docutils literal notranslate"><span class="pre">&gt;&gt;</span></code> and <code class="docutils literal notranslate"><span class="pre">&lt;&lt;</span></code>. The following four statements are all
functionally equivalent:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">op1</span> <span class="o">&gt;&gt;</span> <span class="n">op2</span>
<span class="n">op1</span><span class="o">.</span><span class="n">set_downstream</span><span class="p">(</span><span class="n">op2</span><span class="p">)</span>

<span class="n">op2</span> <span class="o">&lt;&lt;</span> <span class="n">op1</span>
<span class="n">op2</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">(</span><span class="n">op1</span><span class="p">)</span>
</pre>
</div>
</div>
<p>When using the bitshift to compose operators, the relationship is set in the
direction that the bitshift operator points. For example, <code class="docutils literal notranslate"><span class="pre">op1</span> <span class="pre">&gt;&gt;</span> <span class="pre">op2</span></code> means
that <code class="docutils literal notranslate"><span class="pre">op1</span></code> runs first and <code class="docutils literal notranslate"><span class="pre">op2</span></code> runs second. Multiple operators can be
composed &#x2013; keep in mind the chain is executed left-to-right and the rightmost
object is always returned. For example:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">op1</span> <span class="o">&gt;&gt;</span> <span class="n">op2</span> <span class="o">&gt;&gt;</span> <span class="n">op3</span> <span class="o">&lt;&lt;</span> <span class="n">op4</span>
</pre>
</div>
</div>
<p>is equivalent to:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">op1</span><span class="o">.</span><span class="n">set_downstream</span><span class="p">(</span><span class="n">op2</span><span class="p">)</span>
<span class="n">op2</span><span class="o">.</span><span class="n">set_downstream</span><span class="p">(</span><span class="n">op3</span><span class="p">)</span>
<span class="n">op3</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">(</span><span class="n">op4</span><span class="p">)</span>
</pre>
</div>
</div>
<p>For convenience, the bitshift operators can also be used with DAGs. For example:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">dag</span> <span class="o">&gt;&gt;</span> <span class="n">op1</span> <span class="o">&gt;&gt;</span> <span class="n">op2</span>
</pre>
</div>
</div>
<p>is equivalent to:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">op1</span><span class="o">.</span><span class="n">dag</span> <span class="o">=</span> <span class="n">dag</span>
<span class="n">op1</span><span class="o">.</span><span class="n">set_downstream</span><span class="p">(</span><span class="n">op2</span><span class="p">)</span>
</pre>
</div>
</div>
<p>We can put this all together to build a simple pipeline:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="k">with</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;my_dag&apos;</span><span class="p">,</span> <span class="n">start_date</span><span class="o">=</span><span class="n">datetime</span><span class="p">(</span><span class="mi">2016</span><span class="p">,</span> <span class="mi">1</span><span class="p">,</span> <span class="mi">1</span><span class="p">))</span> <span class="k">as</span> <span class="n">dag</span><span class="p">:</span>
    <span class="p">(</span>
        <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;dummy_1&apos;</span><span class="p">)</span>
        <span class="o">&gt;&gt;</span> <span class="n">BashOperator</span><span class="p">(</span>
            <span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;bash_1&apos;</span><span class="p">,</span>
            <span class="n">bash_command</span><span class="o">=</span><span class="s1">&apos;echo &quot;HELLO!&quot;&apos;</span><span class="p">)</span>
        <span class="o">&gt;&gt;</span> <span class="n">PythonOperator</span><span class="p">(</span>
            <span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;python_1&apos;</span><span class="p">,</span>
            <span class="n">python_callable</span><span class="o">=</span><span class="k">lambda</span><span class="p">:</span> <span class="nb">print</span><span class="p">(</span><span class="s2">&quot;GOODBYE!&quot;</span><span class="p">))</span>
    <span class="p">)</span>
</pre>
</div>
</div>
</div>
</div>
<div class="section" id="tasks">
<h3 class="sigil_not_in_toc">Tasks</h3>
<p>Once an operator is instantiated, it is referred to as a &#x201C;task&#x201D;. The
instantiation defines specific values when calling the abstract operator, and
the parameterized task becomes a node in a DAG.</p>
</div>
<div class="section" id="task-instances">
<h3 class="sigil_not_in_toc">Task Instances</h3>
<p>A task instance represents a specific run of a task and is characterized as the
combination of a dag, a task, and a point in time. Task instances also have an
indicative state, which could be &#x201C;running&#x201D;, &#x201C;success&#x201D;, &#x201C;failed&#x201D;, &#x201C;skipped&#x201D;, &#x201C;up
for retry&#x201D;, etc.</p>
</div>
<div class="section" id="workflows">
<h3 class="sigil_not_in_toc">Workflows</h3>
<p>You&#x2019;re now familiar with the core building blocks of Airflow.
Some of the concepts may sound very similar, but the vocabulary can
be conceptualized like this:</p>
<ul class="simple">
<li>DAG: a description of the order in which work should take place</li>
<li>Operator: a class that acts as a template for carrying out some work</li>
<li>Task: a parameterized instance of an operator</li>
<li>Task Instance: a task that 1) has been assigned to a DAG and 2) has a
state associated with a specific run of the DAG</li>
</ul>
<p>By combining <code class="docutils literal notranslate"><span class="pre">DAGs</span></code> and <code class="docutils literal notranslate"><span class="pre">Operators</span></code> to create <code class="docutils literal notranslate"><span class="pre">TaskInstances</span></code>, you can
build complex workflows.</p>
</div>
</div>
<div class="section" id="additional-functionality">
<h2 class="sigil_not_in_toc">Additional Functionality</h2>
<p>In addition to the core Airflow objects, there are a number of more complex
features that enable behaviors like limiting simultaneous access to resources,
cross-communication, conditional execution, and more.</p>
<div class="section" id="hooks">
<h3 class="sigil_not_in_toc">Hooks</h3>
<p>Hooks are interfaces to external platforms and databases like Hive, S3,
MySQL, Postgres, HDFS, and Pig. Hooks implement a common interface when
possible, and act as a building block for operators. They also use
the <code class="docutils literal notranslate"><span class="pre">airflow.models.Connection</span></code> model to retrieve hostnames
and authentication information. Hooks keep authentication code and
information out of pipelines, centralized in the metadata database.</p>
<p>Hooks are also very useful on their own to use in Python scripts,
Airflow airflow.operators.PythonOperator, and in interactive environments
like iPython or Jupyter Notebook.</p>
</div>
<div class="section" id="pools">
<h3 class="sigil_not_in_toc">Pools</h3>
<p>Some systems can get overwhelmed when too many processes hit them at the same
time. Airflow pools can be used to <strong>limit the execution parallelism</strong> on
arbitrary sets of tasks. The list of pools is managed in the UI
(<code class="docutils literal notranslate"><span class="pre">Menu</span> <span class="pre">-&gt;</span> <span class="pre">Admin</span> <span class="pre">-&gt;</span> <span class="pre">Pools</span></code>) by giving the pools a name and assigning
it a number of worker slots. Tasks can then be associated with
one of the existing pools by using the <code class="docutils literal notranslate"><span class="pre">pool</span></code> parameter when
creating tasks (i.e., instantiating operators).</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">aggregate_db_message_job</span> <span class="o">=</span> <span class="n">BashOperator</span><span class="p">(</span>
    <span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;aggregate_db_message_job&apos;</span><span class="p">,</span>
    <span class="n">execution_timeout</span><span class="o">=</span><span class="n">timedelta</span><span class="p">(</span><span class="n">hours</span><span class="o">=</span><span class="mi">3</span><span class="p">),</span>
    <span class="n">pool</span><span class="o">=</span><span class="s1">&apos;ep_data_pipeline_db_msg_agg&apos;</span><span class="p">,</span>
    <span class="n">bash_command</span><span class="o">=</span><span class="n">aggregate_db_message_job_cmd</span><span class="p">,</span>
    <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>
<span class="n">aggregate_db_message_job</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">(</span><span class="n">wait_for_empty_queue</span><span class="p">)</span>
</pre>
</div>
</div>
<p>The <code class="docutils literal notranslate"><span class="pre">pool</span></code> parameter can
be used in conjunction with <code class="docutils literal notranslate"><span class="pre">priority_weight</span></code> to define priorities
in the queue, and which tasks get executed first as slots open up in the
pool. The default <code class="docutils literal notranslate"><span class="pre">priority_weight</span></code> is <code class="docutils literal notranslate"><span class="pre">1</span></code>, and can be bumped to any
number. When sorting the queue to evaluate which task should be executed
next, we use the <code class="docutils literal notranslate"><span class="pre">priority_weight</span></code>, summed up with all of the
<code class="docutils literal notranslate"><span class="pre">priority_weight</span></code> values from tasks downstream from this task. You can
use this to bump a specific important task and the whole path to that task
gets prioritized accordingly.</p>
<p>Tasks will be scheduled as usual while the slots fill up. Once capacity is
reached, runnable tasks get queued and their state will show as such in the
UI. As slots free up, queued tasks start running based on the
<code class="docutils literal notranslate"><span class="pre">priority_weight</span></code> (of the task and its descendants).</p>
<p>Note that by default tasks aren&#x2019;t assigned to any pool and their
execution parallelism is only limited to the executor&#x2019;s setting.</p>
</div>
<div class="section" id="connections">
<span id="concepts-connections"></span><h3 class="sigil_not_in_toc">Connections</h3>
<p>The connection information to external systems is stored in the Airflow
metadata database and managed in the UI (<code class="docutils literal notranslate"><span class="pre">Menu</span> <span class="pre">-&gt;</span> <span class="pre">Admin</span> <span class="pre">-&gt;</span> <span class="pre">Connections</span></code>)
A <code class="docutils literal notranslate"><span class="pre">conn_id</span></code> is defined there and hostname / login / password / schema
information attached to it. Airflow pipelines can simply refer to the
centrally managed <code class="docutils literal notranslate"><span class="pre">conn_id</span></code> without having to hard code any of this
information anywhere.</p>
<p>Many connections with the same <code class="docutils literal notranslate"><span class="pre">conn_id</span></code> can be defined and when that
is the case, and when the <strong>hooks</strong> uses the <code class="docutils literal notranslate"><span class="pre">get_connection</span></code> method
from <code class="docutils literal notranslate"><span class="pre">BaseHook</span></code>, Airflow will choose one connection randomly, allowing
for some basic load balancing and fault tolerance when used in conjunction
with retries.</p>
<p>Airflow also has the ability to reference connections via environment
variables from the operating system. But it only supports URI format. If you
need to specify <code class="docutils literal notranslate"><span class="pre">extra</span></code> for your connection, please use web UI.</p>
<p>If connections with the same <code class="docutils literal notranslate"><span class="pre">conn_id</span></code> are defined in both Airflow metadata
database and environment variables, only the one in environment variables
will be referenced by Airflow (for example, given <code class="docutils literal notranslate"><span class="pre">conn_id</span></code> <code class="docutils literal notranslate"><span class="pre">postgres_master</span></code>,
Airflow will search for <code class="docutils literal notranslate"><span class="pre">AIRFLOW_CONN_POSTGRES_MASTER</span></code>
in environment variables first and directly reference it if found,
before it starts to search in metadata database).</p>
<p>Many hooks have a default <code class="docutils literal notranslate"><span class="pre">conn_id</span></code>, where operators using that hook do not
need to supply an explicit connection ID. For example, the default
<code class="docutils literal notranslate"><span class="pre">conn_id</span></code> for the <a class="reference internal" href="code.html#airflow.hooks.postgres_hook.PostgresHook" title="airflow.hooks.postgres_hook.PostgresHook"><code class="xref py py-class docutils literal notranslate"><span class="pre">PostgresHook</span></code></a> is
<code class="docutils literal notranslate"><span class="pre">postgres_default</span></code>.</p>
<p>See <a class="reference internal" href="howto/manage-connections.html"><span class="doc">Managing Connections</span></a> for how to create and manage connections.</p>
</div>
<div class="section" id="queues">
<h3 class="sigil_not_in_toc">Queues</h3>
<p>When using the CeleryExecutor, the celery queues that tasks are sent to
can be specified. <code class="docutils literal notranslate"><span class="pre">queue</span></code> is an attribute of BaseOperator, so any
task can be assigned to any queue. The default queue for the environment
is defined in the <code class="docutils literal notranslate"><span class="pre">airflow.cfg</span></code>&#x2019;s <code class="docutils literal notranslate"><span class="pre">celery</span> <span class="pre">-&gt;</span> <span class="pre">default_queue</span></code>. This defines
the queue that tasks get assigned to when not specified, as well as which
queue Airflow workers listen to when started.</p>
<p>Workers can listen to one or multiple queues of tasks. When a worker is
started (using the command <code class="docutils literal notranslate"><span class="pre">airflow</span> <span class="pre">worker</span></code>), a set of comma delimited
queue names can be specified (e.g. <code class="docutils literal notranslate"><span class="pre">airflow</span> <span class="pre">worker</span> <span class="pre">-q</span> <span class="pre">spark</span></code>). This worker
will then only pick up tasks wired to the specified queue(s).</p>
<p>This can be useful if you need specialized workers, either from a
resource perspective (for say very lightweight tasks where one worker
could take thousands of tasks without a problem), or from an environment
perspective (you want a worker running from within the Spark cluster
itself because it needs a very specific environment and security rights).</p>
</div>
<div class="section" id="xcoms">
<h3 class="sigil_not_in_toc">XComs</h3>
<p>XComs let tasks exchange messages, allowing more nuanced forms of control and
shared state. The name is an abbreviation of &#x201C;cross-communication&#x201D;. XComs are
principally defined by a key, value, and timestamp, but also track attributes
like the task/DAG that created the XCom and when it should become visible. Any
object that can be pickled can be used as an XCom value, so users should make
sure to use objects of appropriate size.</p>
<p>XComs can be &#x201C;pushed&#x201D; (sent) or &#x201C;pulled&#x201D; (received). When a task pushes an
XCom, it makes it generally available to other tasks. Tasks can push XComs at
any time by calling the <code class="docutils literal notranslate"><span class="pre">xcom_push()</span></code> method. In addition, if a task returns
a value (either from its Operator&#x2019;s <code class="docutils literal notranslate"><span class="pre">execute()</span></code> method, or from a
PythonOperator&#x2019;s <code class="docutils literal notranslate"><span class="pre">python_callable</span></code> function), then an XCom containing that
value is automatically pushed.</p>
<p>Tasks call <code class="docutils literal notranslate"><span class="pre">xcom_pull()</span></code> to retrieve XComs, optionally applying filters
based on criteria like <code class="docutils literal notranslate"><span class="pre">key</span></code>, source <code class="docutils literal notranslate"><span class="pre">task_ids</span></code>, and source <code class="docutils literal notranslate"><span class="pre">dag_id</span></code>. By
default, <code class="docutils literal notranslate"><span class="pre">xcom_pull()</span></code> filters for the keys that are automatically given to
XComs when they are pushed by being returned from execute functions (as
opposed to XComs that are pushed manually).</p>
<p>If <code class="docutils literal notranslate"><span class="pre">xcom_pull</span></code> is passed a single string for <code class="docutils literal notranslate"><span class="pre">task_ids</span></code>, then the most
recent XCom value from that task is returned; if a list of <code class="docutils literal notranslate"><span class="pre">task_ids</span></code> is
passed, then a corresponding list of XCom values is returned.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="c1"># inside a PythonOperator called &apos;pushing_task&apos;</span>
<span class="k">def</span> <span class="nf">push_function</span><span class="p">():</span>
    <span class="k">return</span> <span class="n">value</span>

<span class="c1"># inside another PythonOperator where provide_context=True</span>
<span class="k">def</span> <span class="nf">pull_function</span><span class="p">(</span><span class="o">**</span><span class="n">context</span><span class="p">):</span>
    <span class="n">value</span> <span class="o">=</span> <span class="n">context</span><span class="p">[</span><span class="s1">&apos;task_instance&apos;</span><span class="p">]</span><span class="o">.</span><span class="n">xcom_pull</span><span class="p">(</span><span class="n">task_ids</span><span class="o">=</span><span class="s1">&apos;pushing_task&apos;</span><span class="p">)</span>
</pre>
</div>
</div>
<p>It is also possible to pull XCom directly in a template, here&#x2019;s an example
of what this may look like:</p>
<div class="code sql highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">SELECT</span> <span class="o">*</span> <span class="n">FROM</span> <span class="p">{{</span> <span class="n">task_instance</span><span class="o">.</span><span class="n">xcom_pull</span><span class="p">(</span><span class="n">task_ids</span><span class="o">=</span><span class="s1">&apos;foo&apos;</span><span class="p">,</span> <span class="n">key</span><span class="o">=</span><span class="s1">&apos;table_name&apos;</span><span class="p">)</span> <span class="p">}}</span>
</pre>
</div>
</div>
<p>Note that XComs are similar to <a class="reference internal" href="#variables">Variables</a>, but are specifically designed
for inter-task communication rather than global settings.</p>
</div>
<div class="section" id="variables">
<h3 class="sigil_not_in_toc">Variables</h3>
<p>Variables are a generic way to store and retrieve arbitrary content or
settings as a simple key value store within Airflow. Variables can be
listed, created, updated and deleted from the UI (<code class="docutils literal notranslate"><span class="pre">Admin</span> <span class="pre">-&gt;</span> <span class="pre">Variables</span></code>),
code or CLI. In addition, json settings files can be bulk uploaded through
the UI. While your pipeline code definition and most of your constants
and variables should be defined in code and stored in source control,
it can be useful to have some variables or configuration items
accessible and modifiable through the UI.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="kn">from</span> <span class="nn">airflow.models</span> <span class="k">import</span> <span class="n">Variable</span>
<span class="n">foo</span> <span class="o">=</span> <span class="n">Variable</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s2">&quot;foo&quot;</span><span class="p">)</span>
<span class="n">bar</span> <span class="o">=</span> <span class="n">Variable</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s2">&quot;bar&quot;</span><span class="p">,</span> <span class="n">deserialize_json</span><span class="o">=</span><span class="kc">True</span><span class="p">)</span>
</pre>
</div>
</div>
<p>The second call assumes <code class="docutils literal notranslate"><span class="pre">json</span></code> content and will be deserialized into
<code class="docutils literal notranslate"><span class="pre">bar</span></code>. Note that <code class="docutils literal notranslate"><span class="pre">Variable</span></code> is a sqlalchemy model and can be used
as such.</p>
<p>You can use a variable from a jinja template with the syntax :</p>
<div class="code bash highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">echo</span> <span class="p">{{</span> <span class="n">var</span><span class="o">.</span><span class="n">value</span><span class="o">.&lt;</span><span class="n">variable_name</span><span class="o">&gt;</span> <span class="p">}}</span>
</pre>
</div>
</div>
<p>or if you need to deserialize a json object from the variable :</p>
<div class="code bash highlight-default notranslate"><div class="highlight"><pre><span></span><span class="n">echo</span> <span class="p">{{</span> <span class="n">var</span><span class="o">.</span><span class="n">json</span><span class="o">.&lt;</span><span class="n">variable_name</span><span class="o">&gt;</span> <span class="p">}}</span>
</pre>
</div>
</div>
</div>
<div class="section" id="branching">
<h3 class="sigil_not_in_toc">Branching</h3>
<p>Sometimes you need a workflow to branch, or only go down a certain path
based on an arbitrary condition which is typically related to something
that happened in an upstream task. One way to do this is by using the
<code class="docutils literal notranslate"><span class="pre">BranchPythonOperator</span></code>.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">BranchPythonOperator</span></code> is much like the PythonOperator except that it
expects a python_callable that returns a task_id. The task_id returned
is followed, and all of the other paths are skipped.
The task_id returned by the Python function has to be referencing a task
directly downstream from the BranchPythonOperator task.</p>
<p>Note that using tasks with <code class="docutils literal notranslate"><span class="pre">depends_on_past=True</span></code> downstream from
<code class="docutils literal notranslate"><span class="pre">BranchPythonOperator</span></code> is logically unsound as <code class="docutils literal notranslate"><span class="pre">skipped</span></code> status
will invariably lead to block tasks that depend on their past successes.
<code class="docutils literal notranslate"><span class="pre">skipped</span></code> states propagates where all directly upstream tasks are
<code class="docutils literal notranslate"><span class="pre">skipped</span></code>.</p>
<p>If you want to skip some tasks, keep in mind that you can&#x2019;t have an empty
path, if so make a dummy task.</p>
<p>like this, the dummy task &#x201C;branch_false&#x201D; is skipped</p>
<img alt="https://airflow.apache.org/_images/branch_good.png" src="../img/05acb41b38e78540e05e8e0f1d907a51.jpg">
<p>Not like this, where the join task is skipped</p>
<img alt="https://airflow.apache.org/_images/branch_bad.png" src="../img/fb5803a17d365a3c32b19e03e28a9fde.jpg">
</div>
<div class="section" id="subdags">
<h3 class="sigil_not_in_toc">SubDAGs</h3>
<p>SubDAGs are perfect for repeating patterns. Defining a function that returns a
DAG object is a nice design pattern when using Airflow.</p>
<p>Airbnb uses the <em>stage-check-exchange</em> pattern when loading data. Data is staged
in a temporary table, after which data quality checks are performed against
that table. Once the checks all pass the partition is moved into the production
table.</p>
<p>As another example, consider the following DAG:</p>
<img alt="https://airflow.apache.org/_images/subdag_before.png" src="../img/e9ea586cae938fc2b87189ba6c5cb4f5.jpg">
<p>We can combine all of the parallel <code class="docutils literal notranslate"><span class="pre">task-*</span></code> operators into a single SubDAG,
so that the resulting DAG resembles the following:</p>
<img alt="https://airflow.apache.org/_images/subdag_after.png" src="../img/9231dcec481ea674f2cd8706b9bf499d.jpg">
<p>Note that SubDAG operators should contain a factory method that returns a DAG
object. This will prevent the SubDAG from being treated like a separate DAG in
the main UI. For example:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="c1">#dags/subdag.py</span>
<span class="kn">from</span> <span class="nn">airflow.models</span> <span class="k">import</span> <span class="n">DAG</span>
<span class="kn">from</span> <span class="nn">airflow.operators.dummy_operator</span> <span class="k">import</span> <span class="n">DummyOperator</span>


<span class="c1"># Dag is returned by a factory method</span>
<span class="k">def</span> <span class="nf">sub_dag</span><span class="p">(</span><span class="n">parent_dag_name</span><span class="p">,</span> <span class="n">child_dag_name</span><span class="p">,</span> <span class="n">start_date</span><span class="p">,</span> <span class="n">schedule_interval</span><span class="p">):</span>
  <span class="n">dag</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span>
    <span class="s1">&apos;</span><span class="si">%s</span><span class="s1">.</span><span class="si">%s</span><span class="s1">&apos;</span> <span class="o">%</span> <span class="p">(</span><span class="n">parent_dag_name</span><span class="p">,</span> <span class="n">child_dag_name</span><span class="p">),</span>
    <span class="n">schedule_interval</span><span class="o">=</span><span class="n">schedule_interval</span><span class="p">,</span>
    <span class="n">start_date</span><span class="o">=</span><span class="n">start_date</span><span class="p">,</span>
  <span class="p">)</span>

  <span class="n">dummy_operator</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span>
    <span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;dummy_task&apos;</span><span class="p">,</span>
    <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">,</span>
  <span class="p">)</span>

  <span class="k">return</span> <span class="n">dag</span>
</pre>
</div>
</div>
<p>This SubDAG can then be referenced in your main DAG file:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="c1"># main_dag.py</span>
<span class="kn">from</span> <span class="nn">datetime</span> <span class="k">import</span> <span class="n">datetime</span><span class="p">,</span> <span class="n">timedelta</span>
<span class="kn">from</span> <span class="nn">airflow.models</span> <span class="k">import</span> <span class="n">DAG</span>
<span class="kn">from</span> <span class="nn">airflow.operators.subdag_operator</span> <span class="k">import</span> <span class="n">SubDagOperator</span>
<span class="kn">from</span> <span class="nn">dags.subdag</span> <span class="k">import</span> <span class="n">sub_dag</span>


<span class="n">PARENT_DAG_NAME</span> <span class="o">=</span> <span class="s1">&apos;parent_dag&apos;</span>
<span class="n">CHILD_DAG_NAME</span> <span class="o">=</span> <span class="s1">&apos;child_dag&apos;</span>

<span class="n">main_dag</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span>
  <span class="n">dag_id</span><span class="o">=</span><span class="n">PARENT_DAG_NAME</span><span class="p">,</span>
  <span class="n">schedule_interval</span><span class="o">=</span><span class="n">timedelta</span><span class="p">(</span><span class="n">hours</span><span class="o">=</span><span class="mi">1</span><span class="p">),</span>
  <span class="n">start_date</span><span class="o">=</span><span class="n">datetime</span><span class="p">(</span><span class="mi">2016</span><span class="p">,</span> <span class="mi">1</span><span class="p">,</span> <span class="mi">1</span><span class="p">)</span>
<span class="p">)</span>

<span class="n">sub_dag</span> <span class="o">=</span> <span class="n">SubDagOperator</span><span class="p">(</span>
  <span class="n">subdag</span><span class="o">=</span><span class="n">sub_dag</span><span class="p">(</span><span class="n">PARENT_DAG_NAME</span><span class="p">,</span> <span class="n">CHILD_DAG_NAME</span><span class="p">,</span> <span class="n">main_dag</span><span class="o">.</span><span class="n">start_date</span><span class="p">,</span>
                 <span class="n">main_dag</span><span class="o">.</span><span class="n">schedule_interval</span><span class="p">),</span>
  <span class="n">task_id</span><span class="o">=</span><span class="n">CHILD_DAG_NAME</span><span class="p">,</span>
  <span class="n">dag</span><span class="o">=</span><span class="n">main_dag</span><span class="p">,</span>
<span class="p">)</span>
</pre>
</div>
</div>
<p>You can zoom into a SubDagOperator from the graph view of the main DAG to show
the tasks contained within the SubDAG:</p>
<img alt="https://airflow.apache.org/_images/subdag_zoom.png" src="../img/764cd9d9d35739e2aaba43358950aed5.jpg">
<p>Some other tips when using SubDAGs:</p>
<ul class="simple">
<li>by convention, a SubDAG&#x2019;s <code class="docutils literal notranslate"><span class="pre">dag_id</span></code> should be prefixed by its parent and
a dot. As in <code class="docutils literal notranslate"><span class="pre">parent.child</span></code></li>
<li>share arguments between the main DAG and the SubDAG by passing arguments to
the SubDAG operator (as demonstrated above)</li>
<li>SubDAGs must have a schedule and be enabled. If the SubDAG&#x2019;s schedule is
set to <code class="docutils literal notranslate"><span class="pre">None</span></code> or <code class="docutils literal notranslate"><span class="pre">@once</span></code>, the SubDAG will succeed without having done
anything</li>
<li>clearing a SubDagOperator also clears the state of the tasks within</li>
<li>marking success on a SubDagOperator does not affect the state of the tasks
within</li>
<li>refrain from using <code class="docutils literal notranslate"><span class="pre">depends_on_past=True</span></code> in tasks within the SubDAG as
this can be confusing</li>
<li>it is possible to specify an executor for the SubDAG. It is common to use
the SequentialExecutor if you want to run the SubDAG in-process and
effectively limit its parallelism to one. Using LocalExecutor can be
problematic as it may over-subscribe your worker, running multiple tasks in
a single slot</li>
</ul>
<p>See <code class="docutils literal notranslate"><span class="pre">airflow/example_dags</span></code> for a demonstration.</p>
</div>
<div class="section" id="slas">
<h3 class="sigil_not_in_toc">SLAs</h3>
<p>Service Level Agreements, or time by which a task or DAG should have
succeeded, can be set at a task level as a <code class="docutils literal notranslate"><span class="pre">timedelta</span></code>. If
one or many instances have not succeeded by that time, an alert email is sent
detailing the list of tasks that missed their SLA. The event is also recorded
in the database and made available in the web UI under <code class="docutils literal notranslate"><span class="pre">Browse-&gt;Missed</span> <span class="pre">SLAs</span></code>
where events can be analyzed and documented.</p>
</div>
<div class="section" id="trigger-rules">
<h3 class="sigil_not_in_toc">Trigger Rules</h3>
<p>Though the normal workflow behavior is to trigger tasks when all their
directly upstream tasks have succeeded, Airflow allows for more complex
dependency settings.</p>
<p>All operators have a <code class="docutils literal notranslate"><span class="pre">trigger_rule</span></code> argument which defines the rule by which
the generated task get triggered. The default value for <code class="docutils literal notranslate"><span class="pre">trigger_rule</span></code> is
<code class="docutils literal notranslate"><span class="pre">all_success</span></code> and can be defined as &#x201C;trigger this task when all directly
upstream tasks have succeeded&#x201D;. All other rules described here are based
on direct parent tasks and are values that can be passed to any operator
while creating tasks:</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">all_success</span></code>: (default) all parents have succeeded</li>
<li><code class="docutils literal notranslate"><span class="pre">all_failed</span></code>: all parents are in a <code class="docutils literal notranslate"><span class="pre">failed</span></code> or <code class="docutils literal notranslate"><span class="pre">upstream_failed</span></code> state</li>
<li><code class="docutils literal notranslate"><span class="pre">all_done</span></code>: all parents are done with their execution</li>
<li><code class="docutils literal notranslate"><span class="pre">one_failed</span></code>: fires as soon as at least one parent has failed, it does not wait for all parents to be done</li>
<li><code class="docutils literal notranslate"><span class="pre">one_success</span></code>: fires as soon as at least one parent succeeds, it does not wait for all parents to be done</li>
<li><code class="docutils literal notranslate"><span class="pre">dummy</span></code>: dependencies are just for show, trigger at will</li>
</ul>
<p>Note that these can be used in conjunction with <code class="docutils literal notranslate"><span class="pre">depends_on_past</span></code> (boolean)
that, when set to <code class="docutils literal notranslate"><span class="pre">True</span></code>, keeps a task from getting triggered if the
previous schedule for the task hasn&#x2019;t succeeded.</p>
</div>
<div class="section" id="latest-run-only">
<h3 class="sigil_not_in_toc">Latest Run Only</h3>
<p>Standard workflow behavior involves running a series of tasks for a
particular date/time range. Some workflows, however, perform tasks that
are independent of run time but need to be run on a schedule, much like a
standard cron job. In these cases, backfills or running jobs missed during
a pause just wastes CPU cycles.</p>
<p>For situations like this, you can use the <code class="docutils literal notranslate"><span class="pre">LatestOnlyOperator</span></code> to skip
tasks that are not being run during the most recent scheduled run for a
DAG. The <code class="docutils literal notranslate"><span class="pre">LatestOnlyOperator</span></code> skips all immediate downstream tasks, and
itself, if the time right now is not between its <code class="docutils literal notranslate"><span class="pre">execution_time</span></code> and the
next scheduled <code class="docutils literal notranslate"><span class="pre">execution_time</span></code>.</p>
<p>One must be aware of the interaction between skipped tasks and trigger
rules. Skipped tasks will cascade through trigger rules <code class="docutils literal notranslate"><span class="pre">all_success</span></code>
and <code class="docutils literal notranslate"><span class="pre">all_failed</span></code> but not <code class="docutils literal notranslate"><span class="pre">all_done</span></code>, <code class="docutils literal notranslate"><span class="pre">one_failed</span></code>, <code class="docutils literal notranslate"><span class="pre">one_success</span></code>,
and <code class="docutils literal notranslate"><span class="pre">dummy</span></code>. If you would like to use the <code class="docutils literal notranslate"><span class="pre">LatestOnlyOperator</span></code> with
trigger rules that do not cascade skips, you will need to ensure that the
<code class="docutils literal notranslate"><span class="pre">LatestOnlyOperator</span></code> is <strong>directly</strong> upstream of the task you would like
to skip.</p>
<p>It is possible, through use of trigger rules to mix tasks that should run
in the typical date/time dependent mode and those using the
<code class="docutils literal notranslate"><span class="pre">LatestOnlyOperator</span></code>.</p>
<p>For example, consider the following dag:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="c1">#dags/latest_only_with_trigger.py</span>
<span class="kn">import</span> <span class="nn">datetime</span> <span class="k">as</span> <span class="nn">dt</span>

<span class="kn">from</span> <span class="nn">airflow.models</span> <span class="k">import</span> <span class="n">DAG</span>
<span class="kn">from</span> <span class="nn">airflow.operators.dummy_operator</span> <span class="k">import</span> <span class="n">DummyOperator</span>
<span class="kn">from</span> <span class="nn">airflow.operators.latest_only_operator</span> <span class="k">import</span> <span class="n">LatestOnlyOperator</span>
<span class="kn">from</span> <span class="nn">airflow.utils.trigger_rule</span> <span class="k">import</span> <span class="n">TriggerRule</span>


<span class="n">dag</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span>
    <span class="n">dag_id</span><span class="o">=</span><span class="s1">&apos;latest_only_with_trigger&apos;</span><span class="p">,</span>
    <span class="n">schedule_interval</span><span class="o">=</span><span class="n">dt</span><span class="o">.</span><span class="n">timedelta</span><span class="p">(</span><span class="n">hours</span><span class="o">=</span><span class="mi">4</span><span class="p">),</span>
    <span class="n">start_date</span><span class="o">=</span><span class="n">dt</span><span class="o">.</span><span class="n">datetime</span><span class="p">(</span><span class="mi">2016</span><span class="p">,</span> <span class="mi">9</span><span class="p">,</span> <span class="mi">20</span><span class="p">),</span>
<span class="p">)</span>

<span class="n">latest_only</span> <span class="o">=</span> <span class="n">LatestOnlyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;latest_only&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>

<span class="n">task1</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;task1&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>
<span class="n">task1</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">(</span><span class="n">latest_only</span><span class="p">)</span>

<span class="n">task2</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;task2&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>

<span class="n">task3</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;task3&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>
<span class="n">task3</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">([</span><span class="n">task1</span><span class="p">,</span> <span class="n">task2</span><span class="p">])</span>

<span class="n">task4</span> <span class="o">=</span> <span class="n">DummyOperator</span><span class="p">(</span><span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;task4&apos;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">,</span>
                      <span class="n">trigger_rule</span><span class="o">=</span><span class="n">TriggerRule</span><span class="o">.</span><span class="n">ALL_DONE</span><span class="p">)</span>
<span class="n">task4</span><span class="o">.</span><span class="n">set_upstream</span><span class="p">([</span><span class="n">task1</span><span class="p">,</span> <span class="n">task2</span><span class="p">])</span>
</pre>
</div>
</div>
<p>In the case of this dag, the <code class="docutils literal notranslate"><span class="pre">latest_only</span></code> task will show up as skipped
for all runs except the latest run. <code class="docutils literal notranslate"><span class="pre">task1</span></code> is directly downstream of
<code class="docutils literal notranslate"><span class="pre">latest_only</span></code> and will also skip for all runs except the latest.
<code class="docutils literal notranslate"><span class="pre">task2</span></code> is entirely independent of <code class="docutils literal notranslate"><span class="pre">latest_only</span></code> and will run in all
scheduled periods. <code class="docutils literal notranslate"><span class="pre">task3</span></code> is downstream of <code class="docutils literal notranslate"><span class="pre">task1</span></code> and <code class="docutils literal notranslate"><span class="pre">task2</span></code> and
because of the default <code class="docutils literal notranslate"><span class="pre">trigger_rule</span></code> being <code class="docutils literal notranslate"><span class="pre">all_success</span></code> will receive
a cascaded skip from <code class="docutils literal notranslate"><span class="pre">task1</span></code>. <code class="docutils literal notranslate"><span class="pre">task4</span></code> is downstream of <code class="docutils literal notranslate"><span class="pre">task1</span></code> and
<code class="docutils literal notranslate"><span class="pre">task2</span></code> but since its <code class="docutils literal notranslate"><span class="pre">trigger_rule</span></code> is set to <code class="docutils literal notranslate"><span class="pre">all_done</span></code> it will
trigger as soon as <code class="docutils literal notranslate"><span class="pre">task1</span></code> has been skipped (a valid completion state)
and <code class="docutils literal notranslate"><span class="pre">task2</span></code> has succeeded.</p>
<img alt="https://airflow.apache.org/_images/latest_only_with_trigger.png" src="../img/c93b5f5bd01ebe0b580398d4943a20f3.jpg">
</div>
<div class="section" id="zombies-undeads">
<h3 class="sigil_not_in_toc">Zombies &amp; Undeads</h3>
<p>Task instances die all the time, usually as part of their normal life cycle,
but sometimes unexpectedly.</p>
<p>Zombie tasks are characterized by the absence
of an heartbeat (emitted by the job periodically) and a <code class="docutils literal notranslate"><span class="pre">running</span></code> status
in the database. They can occur when a worker node can&#x2019;t reach the database,
when Airflow processes are killed externally, or when a node gets rebooted
for instance. Zombie killing is performed periodically by the scheduler&#x2019;s
process.</p>
<p>Undead processes are characterized by the existence of a process and a matching
heartbeat, but Airflow isn&#x2019;t aware of this task as <code class="docutils literal notranslate"><span class="pre">running</span></code> in the database.
This mismatch typically occurs as the state of the database is altered,
most likely by deleting rows in the &#x201C;Task Instances&#x201D; view in the UI.
Tasks are instructed to verify their state as part of the heartbeat routine,
and terminate themselves upon figuring out that they are in this &#x201C;undead&#x201D;
state.</p>
</div>
<div class="section" id="cluster-policy">
<h3 class="sigil_not_in_toc">Cluster Policy</h3>
<p>Your local airflow settings file can define a <code class="docutils literal notranslate"><span class="pre">policy</span></code> function that
has the ability to mutate task attributes based on other task or DAG
attributes. It receives a single argument as a reference to task objects,
and is expected to alter its attributes.</p>
<p>For example, this function could apply a specific queue property when
using a specific operator, or enforce a task timeout policy, making sure
that no tasks run for more than 48 hours. Here&#x2019;s an example of what this
may look like inside your <code class="docutils literal notranslate"><span class="pre">airflow_settings.py</span></code>:</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="k">def</span> <span class="nf">policy</span><span class="p">(</span><span class="n">task</span><span class="p">):</span>
    <span class="k">if</span> <span class="n">task</span><span class="o">.</span><span class="vm">__class__</span><span class="o">.</span><span class="vm">__name__</span> <span class="o">==</span> <span class="s1">&apos;HivePartitionSensor&apos;</span><span class="p">:</span>
        <span class="n">task</span><span class="o">.</span><span class="n">queue</span> <span class="o">=</span> <span class="s2">&quot;sensor_queue&quot;</span>
    <span class="k">if</span> <span class="n">task</span><span class="o">.</span><span class="n">timeout</span> <span class="o">&gt;</span> <span class="n">timedelta</span><span class="p">(</span><span class="n">hours</span><span class="o">=</span><span class="mi">48</span><span class="p">):</span>
        <span class="n">task</span><span class="o">.</span><span class="n">timeout</span> <span class="o">=</span> <span class="n">timedelta</span><span class="p">(</span><span class="n">hours</span><span class="o">=</span><span class="mi">48</span><span class="p">)</span>
</pre>
</div>
</div>
</div>
<div class="section" id="documentation-notes">
<h3 class="sigil_not_in_toc">Documentation &amp; Notes</h3>
<p>It&#x2019;s possible to add documentation or notes to your dags &amp; task objects that
become visible in the web interface (&#x201C;Graph View&#x201D; for dags, &#x201C;Task Details&#x201D; for
tasks). There are a set of special task attributes that get rendered as rich
content if defined:</p>
<table border="1" class="docutils">
<colgroup>
<col width="38%">
<col width="62%">
</colgroup>
<thead valign="bottom">
<tr class="row-odd"><th class="head">attribute</th>
<th class="head">rendered to</th>
</tr>
</thead>
<tbody valign="top">
<tr class="row-even"><td>doc</td>
<td>monospace</td>
</tr>
<tr class="row-odd"><td>doc_json</td>
<td>json</td>
</tr>
<tr class="row-even"><td>doc_yaml</td>
<td>yaml</td>
</tr>
<tr class="row-odd"><td>doc_md</td>
<td>markdown</td>
</tr>
<tr class="row-even"><td>doc_rst</td>
<td>reStructuredText</td>
</tr>
</tbody>
</table>
<p>Please note that for dags, doc_md is the only attribute interpreted.</p>
<p>This is especially useful if your tasks are built dynamically from
configuration files, it allows you to expose the configuration that led
to the related tasks in Airflow.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="sd">&quot;&quot;&quot;</span>
<span class="sd">### My great DAG</span>
<span class="sd">&quot;&quot;&quot;</span>

<span class="n">dag</span> <span class="o">=</span> <span class="n">DAG</span><span class="p">(</span><span class="s1">&apos;my_dag&apos;</span><span class="p">,</span> <span class="n">default_args</span><span class="o">=</span><span class="n">default_args</span><span class="p">)</span>
<span class="n">dag</span><span class="o">.</span><span class="n">doc_md</span> <span class="o">=</span> <span class="vm">__doc__</span>

<span class="n">t</span> <span class="o">=</span> <span class="n">BashOperator</span><span class="p">(</span><span class="s2">&quot;foo&quot;</span><span class="p">,</span> <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">)</span>
<span class="n">t</span><span class="o">.</span><span class="n">doc_md</span> <span class="o">=</span> <span class="s2">&quot;&quot;&quot;</span><span class="se">\</span>
<span class="s2">#Title&quot;</span>
<span class="s2">Here&apos;s a [url](www.airbnb.com)</span>
<span class="s2">&quot;&quot;&quot;</span>
</pre>
</div>
</div>
<p>This content will get rendered as markdown respectively in the &#x201C;Graph View&#x201D; and
&#x201C;Task Details&#x201D; pages.</p>
</div>
<div class="section" id="jinja-templating">
<span id="id1"></span><h3 class="sigil_not_in_toc">Jinja Templating</h3>
<p>Airflow leverages the power of
<a class="reference external" href="http://jinja.pocoo.org/docs/dev/">Jinja Templating</a> and this can be a
powerful tool to use in combination with macros (see the <a class="reference internal" href="code.html#macros"><span class="std std-ref">Macros</span></a> section).</p>
<p>For example, say you want to pass the execution date as an environment variable
to a Bash script using the <code class="docutils literal notranslate"><span class="pre">BashOperator</span></code>.</p>
<div class="code python highlight-default notranslate"><div class="highlight"><pre><span></span><span class="c1"># The execution date as YYYY-MM-DD</span>
<span class="n">date</span> <span class="o">=</span> <span class="s2">&quot;{{ ds }}&quot;</span>
<span class="n">t</span> <span class="o">=</span> <span class="n">BashOperator</span><span class="p">(</span>
    <span class="n">task_id</span><span class="o">=</span><span class="s1">&apos;test_env&apos;</span><span class="p">,</span>
    <span class="n">bash_command</span><span class="o">=</span><span class="s1">&apos;/tmp/test.sh &apos;</span><span class="p">,</span>
    <span class="n">dag</span><span class="o">=</span><span class="n">dag</span><span class="p">,</span>
    <span class="n">env</span><span class="o">=</span><span class="p">{</span><span class="s1">&apos;EXECUTION_DATE&apos;</span><span class="p">:</span> <span class="n">date</span><span class="p">})</span>
</pre>
</div>
</div>
<p>Here, <code class="docutils literal notranslate"><span class="pre">{{</span> <span class="pre">ds</span> <span class="pre">}}</span></code> is a macro, and because the <code class="docutils literal notranslate"><span class="pre">env</span></code> parameter of the
<code class="docutils literal notranslate"><span class="pre">BashOperator</span></code> is templated with Jinja, the execution date will be available
as an environment variable named <code class="docutils literal notranslate"><span class="pre">EXECUTION_DATE</span></code> in your Bash script.</p>
<p>You can use Jinja templating with every parameter that is marked as &#x201C;templated&#x201D;
in the documentation. Template substitution occurs just before the pre_execute
function of your operator is called.</p>
</div>
</div>
<div class="section" id="packaged-dags">
<h2 class="sigil_not_in_toc">Packaged dags</h2>
<p>While often you will specify dags in a single <code class="docutils literal notranslate"><span class="pre">.py</span></code> file it might sometimes
be required to combine dag and its dependencies. For example, you might want
to combine several dags together to version them together or you might want
to manage them together or you might need an extra module that is not available
by default on the system you are running airflow on. To allow this you can create
a zip file that contains the dag(s) in the root of the zip file and have the extra
modules unpacked in directories.</p>
<p>For instance you can create a zip file that looks like this:</p>
<div class="highlight-bash notranslate"><div class="highlight"><pre><span></span>my_dag1.py
my_dag2.py
package1/__init__.py
package1/functions.py
</pre>
</div>
</div>
<p>Airflow will scan the zip file and try to load <code class="docutils literal notranslate"><span class="pre">my_dag1.py</span></code> and <code class="docutils literal notranslate"><span class="pre">my_dag2.py</span></code>.
It will not go into subdirectories as these are considered to be potential
packages.</p>
<p>In case you would like to add module dependencies to your DAG you basically would
do the same, but then it is more to use a virtualenv and pip.</p>
<div class="highlight-bash notranslate"><div class="highlight"><pre><span></span>virtualenv zip_dag
<span class="nb">source</span> zip_dag/bin/activate

mkdir zip_dag_contents
<span class="nb">cd</span> zip_dag_contents

pip install --install-option<span class="o">=</span><span class="s2">&quot;--install-lib=</span><span class="nv">$PWD</span><span class="s2">&quot;</span> my_useful_package
cp ~/my_dag.py .

zip -r zip_dag.zip *
</pre>
</div>
</div>
<div class="admonition note">
<p class="first admonition-title">Note</p>
<p class="last">the zip file will be inserted at the beginning of module search list
(sys.path) and as such it will be available to any other code that resides
within the same interpreter.</p>
</div>
<div class="admonition note">
<p class="first admonition-title">Note</p>
<p class="last">packaged dags cannot be used with pickling turned on.</p>
</div>
<div class="admonition note">
<p class="first admonition-title">Note</p>
<p class="last">packaged dags cannot contain dynamic libraries (eg. libz.so) these need
to be available on the system if a module needs those. In other words only
pure python modules can be packaged.</p>
</div>
</div>
</body>
</html>